package com.lzy.elasticserch.result;

import com.lzy.canal.component.result.BaseConsumerResultData;
import com.lzy.canal.component.result.mysql.AbstractConsumerDmlRestData;
import com.lzy.canal.component.result.mysql.DefaultConsumerDdlRestData;
import com.lzy.canal.enums.ConsumerRestTypeEnum;
import com.lzy.canal.model.CanalMessageEntryResultData;
import com.lzy.elasticserch.mapping.DefaultIndexMapping;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;

import java.util.List;
import java.util.stream.Stream;

/**
 * 默认的对象转换器
 *
 */
public class DefaultResultConvert implements ResultConvert {

    private final DefaultIndexMapping indexMapping;

    public DefaultResultConvert(DefaultIndexMapping indexMapping) {
        this.indexMapping = indexMapping;
    }

    @Override
    public DocWriteRequest<?>[] convert(List<CanalMessageEntryResultData> data) {
        return data.stream()
                .flatMap(o ->
                        Stream.of(o.getResults())
                                .filter(j -> !(j instanceof DefaultConsumerDdlRestData))
                                .map(j -> {
                                            DocWriteRequest request = pares(j);
                                            request.index(getIndexToIndexMapping(o.getSchemaName(), o.getTableName()));
                                            return request;
                                        }
                                ))
                .toArray(DocWriteRequest<?>[]::new);
    }


    private DocWriteRequest pares(BaseConsumerResultData result) {
        AbstractConsumerDmlRestData data = (AbstractConsumerDmlRestData) result;
        if (ConsumerRestTypeEnum.INSERT.equals(data.getType())) {
            return new IndexRequest().id(data.getId()).source(data.getResource());
        }
        if (ConsumerRestTypeEnum.UPDATE.equals(data.getType())) {
            return new UpdateRequest().id(data.getId()).doc(data.getResource()).docAsUpsert(true);
        }
        return new DeleteRequest().id(data.getId());
    }

    /**
     * 获取索引
     *
     * @param schema
     * @param table
     * @return
     */
    private String getIndexToIndexMapping(String schema, String table) {
        return indexMapping.mapping(schema, table);
    }
}
